package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.bean.TrafficChIsNewUserSkipBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Classname DwsTrafficUserJumpViewWindow
 * @Description TODO
 * @Date 2023/2/17 18:19
 * @Created by lzx
 */
public class DwsTrafficUserJumpViewWindow {
    public static void main(String[] args) throws Exception {
        /*
        1.环境 2.检查点
        3.从dwd_traffic_user_jump_detail拿数据
        4.格式转换 Str -> JSONObject
        5.提取维度(渠道和新老访客)

         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);


        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new HashMapStateBackend());
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/eduRealTime/ck");
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        DataStreamSource<String> userJumpDS = env.addSource(KafkaUtils.getFlinkKafkaConsumer("dwd_traffic_user_jump_detail", "dws_traffic_user_jump_view_window_group"));
        DataStreamSource<String> pageDS = env.addSource(KafkaUtils.getFlinkKafkaConsumer("dwd_traffic_page_log", "dws_traffic_user_jump_view_window_group"));

        SingleOutputStreamOperator<TrafficChIsNewUserSkipBean> skipDS = userJumpDS.map(new MapFunction<String, TrafficChIsNewUserSkipBean>() {
                    @Override
                    public TrafficChIsNewUserSkipBean map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        String ch = common.getString("ch");
                        String isNew = common.getString("is_new");
                        return new TrafficChIsNewUserSkipBean("", "", ch, isNew, 0L, 1L, jsonObject.getLong("ts"));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficChIsNewUserSkipBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((e, r) -> e.getTs()))
                .keyBy(new KeySelector<TrafficChIsNewUserSkipBean, Tuple2<String,String>>() {
                    @Override
                    public Tuple2<String, String> getKey(TrafficChIsNewUserSkipBean value) throws Exception {
                        return Tuple2.of(value.getCh(), value.getIsNew());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficChIsNewUserSkipBean>() {
                    @Override
                    public TrafficChIsNewUserSkipBean reduce(TrafficChIsNewUserSkipBean value1, TrafficChIsNewUserSkipBean value2) throws Exception {
                        value1.setSkipTotal(value1.getSkipTotal() + value2.getSkipTotal());
                        return value1;
                    }
                }, new ProcessWindowFunction<TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean, Tuple2<String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple2<String, String> stringStringTuple2, ProcessWindowFunction<TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean, Tuple2<String, String>, TimeWindow>.Context context, Iterable<TrafficChIsNewUserSkipBean> elements, Collector<TrafficChIsNewUserSkipBean> out) throws Exception {
                        TrafficChIsNewUserSkipBean next = elements.iterator().next();
                        long end = context.window().getEnd();
                        long start = context.window().getStart();
                        next.setStt(DateFormatUtil.toYmdHms(start));
                        next.setEdt(DateFormatUtil.toYmdHms(end));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });
        //skipDS.print();
        SingleOutputStreamOperator<TrafficChIsNewUserSkipBean> sidTotalDS = pageDS.map(new MapFunction<String, TrafficChIsNewUserSkipBean>() {
                    @Override
                    public TrafficChIsNewUserSkipBean map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        String ch = common.getString("ch");
                        String isNew = common.getString("is_new");
                        return new TrafficChIsNewUserSkipBean("", "", ch, isNew, 1L, 0L, jsonObject.getLong("ts"));
                    }
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficChIsNewUserSkipBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((e, r) -> e.getTs()))
                .keyBy(new KeySelector<TrafficChIsNewUserSkipBean, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(TrafficChIsNewUserSkipBean value) throws Exception {
                        return Tuple2.of(value.getCh(), value.getIsNew());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficChIsNewUserSkipBean>() {
                    @Override
                    public TrafficChIsNewUserSkipBean reduce(TrafficChIsNewUserSkipBean value1, TrafficChIsNewUserSkipBean value2) throws Exception {
                        value1.setSidTotal(value1.getSidTotal() + value2.getSidTotal());
                        return value1;
                    }
                }, new ProcessWindowFunction<TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean, Tuple2<String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple2<String, String> stringStringTuple2, ProcessWindowFunction<TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean, Tuple2<String, String>, TimeWindow>.Context context, Iterable<TrafficChIsNewUserSkipBean> elements, Collector<TrafficChIsNewUserSkipBean> out) throws Exception {
                        TrafficChIsNewUserSkipBean next = elements.iterator().next();
                        long end = context.window().getEnd();
                        long start = context.window().getStart();
                        next.setStt(DateFormatUtil.toYmdHms(start));
                        next.setEdt(DateFormatUtil.toYmdHms(end));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });
        SingleOutputStreamOperator<TrafficChIsNewUserSkipBean> process = skipDS.connect(sidTotalDS)
                .keyBy(TrafficChIsNewUserSkipBean::getStt, TrafficChIsNewUserSkipBean::getStt)
                .process(new CoProcessFunction<TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean>() {
                    private ValueState<TrafficChIsNewUserSkipBean> p1 = null;
                    private ValueState<TrafficChIsNewUserSkipBean> p2 = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<TrafficChIsNewUserSkipBean> p1Des = new ValueStateDescriptor<>("p1", TrafficChIsNewUserSkipBean.class);
                        p1Des.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.seconds(30)).build());
                        ValueStateDescriptor<TrafficChIsNewUserSkipBean> p2Des = new ValueStateDescriptor<>("p2", TrafficChIsNewUserSkipBean.class);
                        p2Des.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.seconds(30)).build());
                        p1 = getRuntimeContext().getState(p1Des);
                        p2 = getRuntimeContext().getState(p2Des);

                    }

                    @Override
                    public void processElement1(TrafficChIsNewUserSkipBean value, CoProcessFunction<TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean>.Context ctx, Collector<TrafficChIsNewUserSkipBean> out) throws Exception {
                        TrafficChIsNewUserSkipBean p2Value = p2.value();
                        if (p2Value == null) {
                            p1.update(value);
                        } else {
                            value.setSidTotal(p2Value.getSidTotal());
                            out.collect(value);
                        }
                    }

                    @Override
                    public void processElement2(TrafficChIsNewUserSkipBean value, CoProcessFunction<TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean, TrafficChIsNewUserSkipBean>.Context ctx, Collector<TrafficChIsNewUserSkipBean> out) throws Exception {
                        TrafficChIsNewUserSkipBean p1Value = p1.value();
                        if (p1Value == null) {
                            p2.update(value);
                        } else {
                            value.setSkipTotal(p1Value.getSkipTotal());
                            out.collect(value);
                        }
                    }
                });
        //process.print();
        process.addSink(ClickHouseUtils.getClickHouseSinkFunction("insert into dws_traffic_user_jump_view_window values(?,?,?,?,?,?,?)"));

        env.execute();
    }
}
